package com.lagou.flashsale.listener;

import com.lagou.flashsale.dao.OrderDAO;
import com.lagou.flashsale.entity.Order;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
@RocketMQMessageListener(topic = "tp_flashsale",
        consumerGroup = "grp_flashsale",
        consumeMode = ConsumeMode.CONCURRENTLY,
        messageModel = MessageModel.CLUSTERING,
        secretKey = "*")
public class RocketMQOrderListener implements RocketMQListener<Order> {

    @Autowired
    private OrderDAO orderDAO;

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Override
    public void onMessage(Order order) {
        System.out.println(order.getOrderStatus().getStatus());
        System.out.println("向数据库插入记录-service - " + order);

        orderDAO.insert(order);

        Message message = new Message("tp_flashpay", order.getOrderId().getBytes());

        org.springframework.messaging.Message springMessage = RocketMQUtil.convertToSpringMessage(message);

        rocketMQTemplate.asyncSend(message.getTopic(), springMessage, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {

            }

            @Override
            public void onException(Throwable throwable) {
            }
        }, 3000, 3);

    }
}
